<?php
namespace ESTool;
/**
 * 链接ES,并获取数据（当成链接池使用）
 * @author xianfe@leju.com
 * @version 0.1 测试
 */
class Elasticsearchpool {
    static $ES_HOST = '127.0.0.1';// = '10.207.9.235';
    static $ES_PORT = '9200';// = '9200';
    static $ES_AUTH_USER;// = '9200';
    static $ES_AUTH_PASSWD;// = '9200';
    static $ES_INDEX = 'access-*';
    
    static public function setESHOST() {
        
        if (isset($_REQUEST['host']) && $_REQUEST['host']=='151') {
            self::$ES_HOST = '172.16.218.96';
            self::$ES_PORT = '19200';
        }  elseif(isset($_REQUEST['host']) && $_REQUEST['host']=='235') {
            self::$ES_HOST = '10.207.6.235';
            self::$ES_PORT = '9200';
            self::$ES_AUTH_USER = 'elastic';
            self::$ES_AUTH_PASSWD = '123456';
        } else {
            $list = array(
//                 "10.204.13.111",
//                 "10.204.13.112",
//                 "10.204.13.113",
                "127.0.0.1",
            );
            self::$ES_PORT = '9200';
            self::$ES_AUTH_USER = 'elastic';
            self::$ES_AUTH_PASSWD = 'admin@123qwe';
            //self::$ES_HOST = '10.204.13.111';
            if (self::$ES_AUTH_USER != '') {
                $auth['user'] = self::$ES_AUTH_USER;
                $auth['passwd'] = self::$ES_AUTH_PASSWD;
            }
            $current_time = time();
            do {
                $index = $current_time % count($list);
                self::$ES_HOST = $list[$index];
                $content = json_decode(self::Curl('', $index='', $mapping='', $node="test", $type=''), true);
                if ($content['cluster_name'] == "club") {
                    break;
                }
                unset($list[$index]);
            } while (count($list)>0);
        }
    }
    /**
     * 获取表结构
     * @param string $index 索引名称
     * @return array 索引结构数据
     */
    static public function getMapping($index='') {
        self::setESHOST();
        $res = self::Curl('', $index, $mapping='', $node="test", $type='_mappings');
        $data = current(json_decode($res, true));
        $data = current($data['mappings']);
        return $data["properties"]["@message"]["properties"];
    }
    /**
     * 保存数据
     * @param array $data 存储数据
     */
    static public function save($data, $index, $mapping='access',$id_filed=array("city", "id"),$limit=3000, $node= "test") {
        // self::setESHOST();
        Debug::op_memory_set(__FILE__, __LINE__, 'Elasticsearchpool_curl');
        $count = 0;
        $post_content = $return = "";
        foreach ($data as $item) {
            ksort($item);
            $id = implode("-", array_intersect_key((array)$item, array_flip($id_filed)));
            $post_content .= '{ "create": { "_index": "'.$index.'", "_type": "'.$mapping.'", "_id": "'.$id.'" }}';
            $post_content .= "\n";
            $post_content .= json_encode($item, JSON_UNESCAPED_UNICODE);
            $post_content .= "\n";
            // 提交数据
            if ($count++ % $limit == 1 || count($data) == $count) {
                $start_time = self::microtime_float();
                $res = self::Curl($post_content, $index, $mapping, $node, $type='_bulk');
                $post_content = "";
                if(strstr($res, '"errors":true')!==false) {
                    $res = json_decode($res, true);
                    foreach ($res["items"] as $err) {
                        $return .= $err["create"]["_id"].":".$err["create"]["status"] ."|";
                    }
                }
                $over_time = '--request_time:' . round((self::microtime_float() - $start_time)*1000) . "ms";
                Debug::op_memory_set(__FILE__, __LINE__, 'Elasticsearchpool_curl', $over_time);
            }
        }
        $return = trim($return, '|');
        return $return;
    }
    /**
     * 更新数据
     * @param array $data 存储数据
     */
    static public function update($data, $index, $mapping='access',$id_filed=array("city", "id"),$limit=3000, $node= "test") {
        // self::setESHOST();
        Debug::op_memory_set(__FILE__, __LINE__, 'Elasticsearchpool_curl');
        $count = 0;
        $post_content = $return = "";
        foreach ($data as $item) {
            ksort($item);
            $id = implode("-", array_intersect_key((array)$item, array_flip($id_filed)));
            $post_content .= '{ "update": { "_index": "'.$index.'", "_type": "'.$mapping.'", "_id": "'.$id.'" }}';
            $post_content .= "\n";
            $post_content .= '{"doc": '.json_encode($item, JSON_UNESCAPED_UNICODE) . '}';
            $post_content .= "\n";
            // 提交数据
            if ($count++ % $limit == 1 || count($data) == $count) {
                $start_time = self::microtime_float();
                $res = self::Curl($post_content, $index, $mapping, $node, $type='_bulk');
                $post_content = "";
                if(strstr($res, '"errors":true')!==false) {
                    $res = json_decode($res, true);
                    foreach ($res["items"] as $err) {
                        $return .= $err["update"]["_id"].":".$err["update"]["status"] ."|";
                    }
                }
                $over_time = '--request_time:' . round((self::microtime_float() - $start_time)*1000) . "ms";
                Debug::op_memory_set(__FILE__, __LINE__, 'Elasticsearchpool_curl', $over_time);
            }
        }
        $return = trim($return, '|');
        return $return;
    }
    static public function _getAuth() {
        $auth = array();
        if (self::$ES_AUTH_USER != '') {
            $auth['user'] = self::$ES_AUTH_USER;
            $auth['passwd'] = self::$ES_AUTH_PASSWD;
        }
        return $auth;
    }
    /**
     * 获取es数据
     * @param array $where 查询条件
	 * @param array $group 聚合字段
	 * @param string $time 聚合时间间隔
     * @param string $type mapping 类型
     * @param String $node 请求数据节点
     * @return array
     */
	static public function FetchSourceCount($dsl, $index='', $mapping='pc', $node= "test") {
	    $content = self::Curl($dsl, $index, $mapping, $node, $type= '_search');
		return $content;
	}
    static function convert($size) {
        $unit=array('b','kb','mb','gb','tb','pb');
        return @round($size/pow(1024,($i=floor(log($size,1024)))),2).' '.$unit[$i];
    }
    
    static function getSize($a) {
        $a = strtolower($a);
        if (strstr($a, "k")!==false) {
            return floor($a) * 1024;
        } elseif (strstr($a, "m")!==false) {
            return floor($a) * 1024 * 1024;
        } elseif (strstr($a, "g")!==false) {
            return floor($a) * 1024 * 1024 * 1024;
        } else {
            return floor($a);
        }
        return $a;
    }
    static public function Curl($dsl, $index='', $mapping='', $node= "test", $type= '_search') {
        Debug::op_memory_set(__FILE__, __LINE__, __CLASS__ . __FUNCTION__);
        try {
            // URL拼装
            // self::setESHOST();
            $url = 'http://'.self::$ES_HOST.':'.self::$ES_PORT.'/'.$index;
            $url = array(
                'fomat' => "%s://%s:%s/%s/%s/%s",
                'scheme' => "http",
                'host' => self::$ES_HOST,
                'port' => self::$ES_PORT,
                'index' => empty($index) ? self::$ES_INDEX : $index,
                'mapping' => $mapping,
                'type' => $type,
            );
            // 设置size
            if (is_array($dsl) && !isset($dsl['size'])) {
                $dsl['size'] =  0;
            }
            if (isset($_GET['debug_stat']) && $_GET['debug_stat']=="start") {
                $request = array(
                    __FILE__, __LINE__,
                    "[Request Url]\n".$url,
                    "[Request Body]\n".(is_array($dsl) ?json_encode($dsl):$dsl),
                    "[Node]\n".$node
                );
                p($request); exit;
            }
            // 兼容本地框架，有node节点数据是请求 对应 请求方法
            
            $start_time = self::microtime_float();
            $auth = self::_getAuth();
            if (is_array($dsl)) {
                $dsl = json_encode($dsl);
            }
            $requestURI = call_user_func_array('sprintf', $url);
            $content = self::_Curl($requestURI, $dsl, $auth);
            $over_time = '--request_time:' . round((self::microtime_float() - $start_time)*1000) . "ms";
            $over_time .= '--es_search_time:' . substr($content, 8, strpos($content, ",") -8) . "ms";
            Debug::op_memory_set(__FILE__, __LINE__, __CLASS__ . __FUNCTION__, $over_time);
            $request = array(
                __FILE__, __LINE__,
                "[Request Url]\n".$requestURI,
                "[Request Body]\n". $dsl,
                "[Request Auth]\n".json_encode($auth),
                "[time]\n".$over_time,
                "[Node]\n".$node
            );
            if ((isset($_GET['debug_stat']) && $_GET['debug_stat']=="1q2w3e")) {
                p($request);
            }
            Debug::op_memory_set_query($request);
            return $content;
        } catch(\Exception $e) {
            Debug::op_memory_set_query(__FILE__, __LINE__, "[Request Url]\n".$url, "[Request Body]\n".$dsl, "[time]\n".$over_time);
            echo '<pre class="error" style="background-color: #ccc;font-weight: 600;padding:20px;border-radius: 10px;border:2px solid red">';
            echo "--\n" . 'Elasticsearch Exception : ' . "\n";
            p(array(
                __FILE__, __LINE__,
                "[Request Url]\n".$requestURI,
                "[Request Body]\n". $dsl,
                "[Request Auth]\n".json_encode($auth),
                "[time]\n".$over_time,
                "[Node]\n".$node,
                "[Error]"=>$e->getCode(),
                "[File]"=> __FILE__ . ' ' . $e->getLine(),
                "[Console]"=>$e->getMessage()
            ));
            echo '</pre>';
            exit;
            return false;
        }
    }
    static public function _Curl($url, $postFields = null, $auth='') {
        
		$ch = curl_init();
		if ($auth) {
		    if (is_array($auth)) {
		        $auth = $auth['user'].":".$auth['passwd'];
		    }
		    curl_setopt($ch, CURLOPT_USERPWD, $auth);
		}
		// curl_setopt($ch, CURLOPT_FTPPORT, '10.207.9.253');
		curl_setopt($ch, CURLOPT_URL, $url);
		curl_setopt($ch, CURLOPT_USERAGENT, 'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:45.0) Gecko/20100101 Firefox/45.0');
		curl_setopt($ch, CURLOPT_HEADER, 0);
		curl_setopt($ch, CURLOPT_FAILONERROR, 0);
		curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
		curl_setopt($ch, CURLOPT_ENCODING, 'gzip');
		//curl_setopt($ch,CURLOPT_HTTPHEADER,array("Expect:"));
		if (!empty($postFields)) {
			
			$postBodyString = "";
			$postMultipart = false;
			
			if (is_array($postFields) && 0 < count($postFields)) {
				foreach ($postFields as $k => $v)
				{
					if("@" != substr($v, 0, 1))//判断是不是文件上传
					{
						$postBodyString .= "$k=" . urlencode($v) . "&";
					}
					else//文件上传用multipart/form-data，否则用www-form-urlencoded
					{
						$postMultipart = true;
					}
				}
				unset($k, $v);
			} else {
				$postBodyString = $postFields . "&";
			}
			curl_setopt($ch, CURLOPT_POST, 1);
			if ($postMultipart)
			{
				curl_setopt($ch, CURLOPT_POSTFIELDS, $postFields);
			}
			else
			{
// 				var_dump($postBodyString);
				curl_setopt($ch, CURLOPT_POSTFIELDS, substr($postBodyString,0,-1));
			}
		}
		$reponse = curl_exec($ch);
// 		echo "<pre>";
	// 	var_dump(curl_getinfo($ch));
		//return curl_getinfo($ch);
		if (curl_errno($ch))
		{
			throw new \Exception(curl_error($ch),0);
		}
		else
		{
			$httpStatusCode = curl_getinfo($ch, CURLINFO_HTTP_CODE);
			if (200 !== $httpStatusCode)
			{
				throw new \Exception($reponse,$httpStatusCode);
			}
		}
		curl_close($ch);
		return $reponse;
	}
	
	static public function microtime_float(){
	    list ($usec, $sec) = explode(" ", microtime());
	    return ((float) $usec + (float) $sec);
	}
}